package com.pw.study.flink.window

import com.pw.study.flink.source.MyWordSource
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object W2 {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val ds: DataStream[String] = env.addSource(new MyWordSource())
    val word: DataStream[(String, Int)] = ds.map((_, 1))
    val wordAndOne: KeyedStream[(String, Int), String] = word.keyBy(_._1)
    val dsProcess: DataStream[(String, Int)] = wordAndOne.process(new MyKeyedProcessFunction())

    dsProcess.print("kk: ")
    env.execute("jobManagerName")

  }
}

class MyKeyedProcessFunction extends KeyedProcessFunction[String, (String, Int), (String, Int)] {
  override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#Context, collector: Collector[(String, Int)]): Unit = {
    println("开始学习：key -> " + ctx.getCurrentKey + " 数据：" + value._1 + " --> " + value._2)
    collector.collect(value)
  }
}
